package com.orderpay_detect;

import com.orderpay_detect.bean.OrderEvent;
import com.orderpay_detect.bean.ReceiptEvent;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2021/11/12 17:37
 *
 *  普通方式
 * 两条流的订单交易匹配
 */
public class TxPayMatch {

    //定义侧输出流标签
    private final static OutputTag<OrderEvent> unmatchedPays = new OutputTag<OrderEvent>("unmatched-pays") {
    };

    private final static OutputTag<ReceiptEvent> unmatchedReceipts = new OutputTag<ReceiptEvent>("unmatched-receipts") {
    };

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //从文件中读取数据
        String orderPath = "OrderPayDetect/src/main/resources/OrderLog.csv";
        DataStream<String> orderStream = env.readTextFile(orderPath);

        String receiptPath = "OrderPayDetect/src/main/resources/ReceiptLog.csv";
        DataStream<String> receiptStream = env.readTextFile(receiptPath);

        //将读取的数据转换为POJO类型
        //读取订单支付事件数据
        DataStream<OrderEvent> orderEventStream = orderStream
                .map(line -> {
                    String[] fields = line.split(",");
                    return new OrderEvent(new Long(fields[0]), fields[1], fields[2], new Long(fields[3]));
                })
                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<OrderEvent>() {
                    @Override
                    public long extractAscendingTimestamp(OrderEvent orderEvent) {
                        return orderEvent.getTimestamp() * 1000L;
                    }
                })
                .filter(data -> !"".equals(data.getTxId())); //交易id不为空，必须是pay事件


        //读取到账事件数据
        DataStream<ReceiptEvent> receiptEventStream = receiptStream
                .map(line -> {
                    String[] fields = line.split(",");
                    return new ReceiptEvent(fields[0], fields[1], new Long(fields[2]));
                })
                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<ReceiptEvent>() {
                    @Override
                    public long extractAscendingTimestamp(ReceiptEvent receiptEvent) {
                        return receiptEvent.getTimestamp() * 1000L;
                    }
                });


        //将两条流进行连接合并，进行匹配处理。不匹配事件输出到侧输出流
        SingleOutputStreamOperator<Tuple2<OrderEvent, ReceiptEvent>> resultStream = orderEventStream
                .keyBy(OrderEvent::getTxId)
                .connect(receiptEventStream.keyBy(ReceiptEvent::getTxId))
                .process(new TxPayMatchDetect());


        resultStream.print("matched-pays");
        resultStream.getSideOutput(unmatchedPays).print("unmatched-pays");
        resultStream.getSideOutput(unmatchedReceipts).print("unmatched-receipts");


        env.execute("tx match detect job");
    }

    /**
     * 实现自定义CoProcessFunction函数
     */
    public static class TxPayMatchDetect extends CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>> {

        //定义状态，保存当前已经到来的订单支付事件和到账时间
        ValueState<OrderEvent> payState;
        ValueState<ReceiptEvent> receiptState;

        @Override
        public void open(Configuration parameters) throws Exception {
            payState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("pay", OrderEvent.class));

            receiptState = getRuntimeContext().getState(new ValueStateDescriptor<ReceiptEvent>("receipt", ReceiptEvent.class));
        }

        @Override
        public void processElement1(OrderEvent orderEvent, Context context, Collector<Tuple2<OrderEvent, ReceiptEvent>> collector) throws Exception {

            //订单支付事件来了，判断是否已经有对应的到账事件
            ReceiptEvent receipt = receiptState.value();
            if (receipt != null) {
                //如果receipt不为空，说明到账事件已经来过，输出匹配事件，清空状态
                collector.collect(new Tuple2<>(orderEvent, receipt));
                payState.clear();
                receiptState.clear();
            } else {
                //如果receipt没来，注册一个定时器，开始等待
                context.timerService().registerEventTimeTimer((orderEvent.getTimestamp() + 5) * 1000L); //等待5s
                //更新状态
                payState.update(orderEvent);
            }
        }

        @Override
        public void processElement2(ReceiptEvent receiptEvent, Context context, Collector<Tuple2<OrderEvent, ReceiptEvent>> collector) throws Exception {
            //到账事件来了，判断是否已经有对应的支付事件
            OrderEvent pay = payState.value();
            if (pay != null) {
                //如果pay不为空，说明支付事件已经来过，输出匹配事件，清空状态
                collector.collect(new Tuple2<>(pay, receiptEvent));
                payState.clear();
                receiptState.clear();
            } else {
                //如果pay没来，注册一个定时器，开始等待
                context.timerService().registerEventTimeTimer((receiptEvent.getTimestamp() + 3) * 1000L); //等待3s
                //更新状态
                receiptState.update(receiptEvent);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
            //定时器触发，有可能是有一个事件没来，不匹配，也有可能是都过来了，已经输出并清空了
            //判断那个不为空，那么另一个就没来
            if(payState.value()!=null){
                ctx.output(unmatchedPays,payState.value());
            }
            if (receiptState.value()!=null){
                ctx.output(unmatchedReceipts,receiptState.value());
            }
            //清空状态
            payState.clear();
            receiptState.clear();
        }
    }
}